package com.yunhe.marketing.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.yunhe.common.util.StringUtil;
import com.yunhe.marketing.constant.CompanyEventConsts;
import com.yunhe.marketing.domain.CompanyEvent;
import com.yunhe.marketing.domain.EventMsg4Kafka;
import com.yunhe.marketing.repository.CompanyEventRepository;
import com.yunhe.marketing.service.ICompanyEventService;
import com.yunhe.common.util.CollectionUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;

/**
 * @Author: DongWenYu
 * @Date: 2021/3/12 18:01
 */
@Service
public class CompanyEventServiceImpl implements ICompanyEventService {

    @Autowired
    CompanyEventRepository companyEventRepository;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public void exposureBatch(List<CompanyEvent> companyEvents, Long userId, String ip, String ipAddress) {
        for (CompanyEvent companyEvent : companyEvents){
            companyEvent.setUserId(userId);
            companyEvent.setUserIp(ip);
            companyEvent.setUserLocation(ipAddress);
            if (StringUtil.isEmpty(companyEvent.getEventType())){
                companyEvent.setEventType(CompanyEventConsts.EVENT_TYPE.Exposure.value());
            }
                companyEvent.setRtime(System.currentTimeMillis());
        }
        companyEventRepository.createBatch(companyEvents);

        //发送消息到kafka
        sendMsgToKafka(companyEvents, null);
    }

    @Override
    public void click(CompanyEvent companyEvent, Long userId , String ip, String ipAddress) {
        companyEvent.setUserId(userId);
        companyEvent.setUserIp(ip);
        companyEvent.setUserLocation(ipAddress);
        if (StringUtil.isEmpty(companyEvent.getEventType())){
            companyEvent.setEventType(CompanyEventConsts.EVENT_TYPE.Click.value());
        }
        companyEvent.setRtime(System.currentTimeMillis());
        companyEventRepository.create(companyEvent);

        //发送消息到kafka
        sendMsgToKafka(null, companyEvent);
    }

    /**
     * 发送事件到kafka
     * @param companyEventList 事件列表
     * @param companyEvent  单次事件
     */
    private void sendMsgToKafka(List<CompanyEvent> companyEventList, CompanyEvent companyEvent) {
        EventMsg4Kafka eventMsg4Kafka = new EventMsg4Kafka();
        if (CollectionUtil.isNotEmpty(companyEventList)) {
            eventMsg4Kafka.setEventType(companyEventList.get(0).getEventType());
            eventMsg4Kafka.setEventList(companyEventList);
            kafkaTemplate.send("EventMsg", JSONObject.toJSONString(eventMsg4Kafka));
        } else if (companyEvent != null) {
            eventMsg4Kafka.setEventType(companyEvent.getEventType());
            List<CompanyEvent> newCompanyEventList = new ArrayList<>();
            newCompanyEventList.add(companyEvent);
            eventMsg4Kafka.setEventList(newCompanyEventList);
            kafkaTemplate.send("EventMsg", JSONObject.toJSONString(eventMsg4Kafka));
        }
    }
}
